package com.yuninglong.cdc_demo1;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
/**
 * @author: yuninglong
 */
public class CdcFromMysql {
    public static void main(String[] args) throws Exception {
        new CdcFromMysql().execute(args);
    }

    private void execute(String[] args) throws Exception {

        try {
            //TODO 1）初始化flink流处理的运行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 使用MySQLSource创建数据源
// 同时指定StringDebeziumDeserializationSchema，将CDC转换为String类型输出
            String hostname = "192.168.202.10";
            int port = 3301;
            String database = "demo";
            String username = "root";
            String password = "testYnlcom";
//            StringDebeziumDeserializationSchema deserializer = new StringDebeziumDeserializationSchema();

//            SourceFunction<String> sourceFunction = MySqlSource.<String>builder()
//                    .hostname(hostname).port(port)
//                    .databaseList(database).username(username).password(password)
//                    .deserializer(deserializer)
//                    .build();
// 单并行度打印，避免输出乱序
//            env.addSource(mySqlSource).print().setParallelism(1);

            MySqlSource<String> mySqlSource= MySqlSource.<String>builder().hostname(hostname).port(port)
                    .databaseList(database).username(username).password(password)
                    .tableList("demo.student")
                    .serverId("1-100")
                    .deserializer(new JsonDebeziumDeserializationSchema())
                    .includeSchemaChanges(true) // output the schema changes as well
                    .build();

            env.enableCheckpointing(3000);

// set the source parallelism to 1
            env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySqlParallelSource")
//                    .setParallelism(1)
                    .print()
                    .setParallelism(1);


            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
